package filter.matadata;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.StringTokenizer;
/**
 * Created by luoqifei on 16-9-9.
 *
 */
public class GetColumnAndData extends Configured implements Tool{
    public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private final static IntWritable zero = new IntWritable(0);
        private Text word = new Text();
        public void map(LongWritable key,Text value,Context context) {
            if (!isColumn(value)) {//if the line start with '-',means it was not column name
                String line = value.toString();

                String[] datas = line.split(";");
                for (int i=0; i<datas.length; i++){
                    word.set(datas[i]);
                    try {
                        context.write(word,zero);
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            } else {
                //read line,but replace all of ";"
                String line = value.toString().replaceAll(";", "");
                //split the line by space
                //读取每一行数据，并将该行数据以空格分割（StringTokenizer默认是以空格分割字符串）
                StringTokenizer tokenizer = new StringTokenizer(line);
                while (tokenizer.hasMoreTokens()) {
                    word.set(tokenizer.nextToken());
                    try {
                        context.write(word, one);//输出给Reduce
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    private static  boolean isColumn(Text value){
        if(value.toString().startsWith("-")){
            return false;
        }else {
            return true;
        }
    }
    public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{
            int sum = 0;
            //合并每个单词出现的次数
            for (IntWritable val : values) {
                sum+=val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }
    public static void main(String[] args) throws Exception {
        int ret = ToolRunner.run(new GetColumnAndData(), args);
        System.exit(ret);
    }
    @Override
    public int run(String[] arg0) throws Exception {
        Job job = new Job(getConf());
        job.setJobName("get-column-name");

        job.setOutputKeyClass(Text.class);//key--设置输出格式
        job.setOutputValueClass(IntWritable.class);//value--设置输出格式

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.setInputPaths(job, new Path(arg0[0]));
        FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
        boolean success = job.waitForCompletion(true);

        return success?0:1;
    }
}
